package com.baishancloud.log.report

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.common.source.SourceUtil
import com.baishancloud.log.common.util.{StringUtil, TimeUtil}
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.DataStream


class IcbcReportor extends Serializable

/**
 *
 * @author ziqiang.wang 
 * @date 2021/11/18 11:41
 */
object IcbcReportor extends Serializable {


  val sinkParallel: String = "sinkParallel"


  def main(args: Array[String]): Unit = {
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    println("输入参数：" + parameterTool.toMap.toString)
    paramCheck(parameterTool)
    StreamEnv.builder(parameterTool).build()
    //    val source: DataStream[String] = SourceUtil.fileOnlyValue(parameterTool)
    val source1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "1").uid("75e77822-0818-4548-a6a7-6ace1ace2bc8")
    val source2: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "2").uid("2175206e-ded6-41b0-9dff-66b45bb2a905")
    val source: DataStream[String] = source1.union(source2)
    source
      .map(Origin.parse(_).orNull).name("Origin").uid("0465a7e2-7932-4257-aac2-cb037c4b05d4")
      .filter(_ != null).name("!=null").uid("ed43ee8d-fc6f-4cfe-9c1b-40b2ee2adc0a")
      .filter(new DomainIdFilter(parameterTool)).name("DomainIdFilter").uid("5ac2ec35-0b8c-4a13-82e0-a7580fa9b9b5")
      .map(o => Log(StringUtil.subString(o.domain(), 100), o.ip(), o.httpCode, o.serverIp, TimeUtil.longToString(o.datetime.toEpochSecond), o.hit, o.hierarchy)).name("toLogObject").uid("305a65e2-97d4-4580-9370-f0868b1dcf78")
      .map(JSONUtil.toJsonStr(_)).name("toJsonStr").uid("f9963cc2-aa89-41b2-b85a-59d194776797")
      .addSink(SinkUtil.starRocksJsonString(parameterTool)).setParallelism(parameterTool.getInt(sinkParallel, 1)).name("statis_icbc_origin_errors").uid("221bb73a-a59d-4ed9-bc59-e6f86b612c96")
    //      .print("结果数据：")

    StreamEnv.execute(parameterTool)
  }

  /**
   * 主类输入参数检查
   */
  private def paramCheck(parameterTool: ParameterTool): Unit = {
    if (!parameterTool.has(accounts)) {
      throw new IllegalArgumentException(s"未设置 --${accounts}  参数")
    }
  }

}
